package com.aotain.coeus.mr;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;


/*
 * spark-submit --class com.aotain.project.mr.ProbaSpark --deploy-mode cluster --master yarn --num-executors 8 --driver-memory 4g --executor-memory 2g --executor-cores 1 --queue thequeue /home/bigdata/project/model/bin/probamodel.jar /user/hive/warehouse/tianwang.db/probamodel/name=baidu.com/ /user/hive/warehouse/tianwang.db/host/name=baidu.com/  /user/hive/warehouse/tianwang.db/urlabnormodel/name=baidu.com/ 
 * 
 * 
 * 
 * */

public class ProbaSpark {

	public static void main(String[] args) {		
		
		if (args.length != 3){
            System.err.printf("Usage: %s <output><input><config>");
            System.exit(1);
		}
		ProbaFunc(args);		 
	}
	
	 public static int ProbaFunc(final String[] args){
		 try
		 {
		    SparkConf sparkConf = new SparkConf().setAppName("Proba_Spark");
		    Configuration config = new Configuration();
		    config.addResource("/etc/hadoop/conf");
		    JavaSparkContext ctx = new JavaSparkContext(sparkConf);
		    
		    String sourcePath = args[1];
		    System.out.println("sourcePath: " + sourcePath);

		    String targetPath = args[0];
		    System.out.println("targetPath: " + targetPath);
		    FileSystem fsTarget = FileSystem.get(URI.create(targetPath),config);
		    Path pathTarget = new Path(targetPath);
		    if(fsTarget.exists(pathTarget))
		    {
		    	fsTarget.delete(pathTarget, true);
		    	System.out.println("Delete path " + targetPath);
		    }
		    
		    String confuri = args[2]; 
		    FileSystem fs = FileSystem.get(URI.create(confuri),config); 
            FSDataInputStream in = null;
            final HashMap<String, HashMap<String, Long>> mappath = new HashMap<String, HashMap<String, Long>>();
            final HashMap<String, HashMap<String, Long>> mappara = new HashMap<String, HashMap<String, Long>>();
            Path path = new Path(confuri);
            if(fs.exists(path))
            {
            	System.out.println(confuri); 
            	System.out.println("exist conf file !!!!!!"); 
            	try{
            		if(fs.isDirectory(path))
            		{            		
            			for(FileStatus file:fs.listStatus(path))
            			{
			           		try{
			           			in = fs.open(file.getPath());
			    	           	BufferedReader bis = new BufferedReader(new InputStreamReader(in,"UTF8")); 
			    	           	String line = "";
			    	           	while ((line = bis.readLine()) != null) {  
			    	           		String items[] = line.split("\\|",-1);
			    	           		if(items[1].equals("1")){ //path model
			    	           			HashMap<String, Long> pps = null;	
				    		        	if(mappath.containsKey(items[0])){	        		
				    		        		pps = mappath.get(items[0]);
				    		        		pps.put(items[2],Long.parseLong(items[3].trim()));
				    		        	}else{
				    		        		pps = new HashMap<String, Long>();
				    		        		pps.put(items[2],Long.parseLong(items[3].trim()));
				    		        		mappath.put(items[0], pps);	
				    		        	}
			    	           		}else if(items[1].equals("2")){//paras model
			    	           			HashMap<String, Long> pas = null;	
				    		        	if(mappara.containsKey(items[0])){	        		
				    		        		pas = mappara.get(items[0]);
				    		        		pas.put(items[2],Long.parseLong(items[3].trim()));
				    		        	}else{
				    		        		pas = new HashMap<String, Long>();
				    		        		pas.put(items[2],Long.parseLong(items[3].trim()));
				    		        		mappara.put(items[0], pas);	
				    		        	}
			    	           		}
			    	           	}				            
			           		}finally{
			           			IOUtils.closeStream(in);
			           		}
            			}
            		}
            	}finally{ 
                    IOUtils.closeStream(in); 
            	}
            }
        	else
        	{
        		System.out.println(confuri); 
        		System.out.println("not exist file !!!!!!"); 
        	}
            JavaPairRDD<LongWritable,Text> tt=ctx.newAPIHadoopFile(sourcePath, TextInputFormat.class, LongWritable.class, Text.class,config);

		    JavaPairRDD<String, String> s = tt.mapToPair(new PairFunction<Tuple2<LongWritable,Text>, String,String>() {
		    	
		    	/**
				 * a.hiphotos.baidu.com|YS5oaXBob3Rvcy5iYWlkdS5j
				 * domain|url
				 */
				private static final long serialVersionUID = 1L;
				
				@Override
				public Tuple2<String, String> call(Tuple2<LongWritable, Text> v1) throws Exception {
					
					String row = v1._2.toString();
					String items[]  = row.split("\\|", -1);
				
					Tuple2<String, String> result = new Tuple2<String, String>("", "");					 
					if(items[0].equals("") || items[1].equals(""))
						 return result;
										 
					String url = new String(Base64.decodeBase64(items[1])).toLowerCase();
					String SuffixName = "cgi|tpt|mp4|htc|bif|m4a|so|ipcc|ipd|patch|trt|cfg|torrent|crl|lcr|fcg|do|m3u8|config|cdr|mkv|tsd|crc|flv|wav|manifest|mpq|mfil|swz|dns|msi|pack|ts|ver|ctl|srf|zip|rar|bmp|ico|pic|exe|doc|docx|pdf|xls|xlsx|gz|f4v|gif|jpg|png|css|json|js|txt|eot|id|gzip|cab|7z|vbs|cer|etf|kwp|swf|dat|ini|dll|zl|msu|lua|asa|psf|jpeg|qpyd|blf|ads|bin|xdelta|webp|xml|lrc|data|e|lst|npk|crt|rp|dtd|msp|rmvb|bup|sdo|org|cit|qgi";
					String[] sfarr = SuffixName.split("\\|",-1);
					for(String sf : sfarr){
						if(url.contains("."+sf)) return result;
					}
					
					if(url.contains("://")){
						url = url.split("://",-1)[1];
					}
					double proba1 = 1.00;
					double proba2 = 1.00;
					//web.api.muyingzhijia.com/api/responsedynamicadvertis?callback=jquery1820903160450406802_1450762833507&_=1450762833840"
					String[] urls = url.split("/",-1);
					int loop = urls.length > 3 ? 3 : urls.length-1;
					for(int i = 1; i < loop; i++){
						if(urls[i-1].isEmpty() || urls[i].isEmpty()) break;
						String k1 = urls[i-1] + "/" + urls[i];
						if(mappath.containsKey(items[0])){
							HashMap<String, Long> ls = mappath.get(items[0]);
							if(ls.containsKey(urls[i-1]) && ls.containsKey(k1)){
								proba1 = ((double)ls.get(k1) / (double)ls.get(urls[i-1])) * proba1;
							}
						}
					}
//					if(url.contains("?") && mappara.containsKey(items[0])){
//						Map<String, String> mapRequest = CRequest.URLRequest(url);
//						for(String strRequestKey: mapRequest.keySet()) {
//							String k1 = strRequestKey;
//							String k2 = CRequest.UrlPage(url)+"?"+strRequestKey;
//							
//							HashMap<String, Long> ls = mappara.get(items[0]);
//							if(ls.containsKey(k1) && ls.containsKey(k2)){
//								proba2 = ((double)ls.get(k1) / (double)ls.get(k2)) * proba2;	
//							}
//						}
//					}
					if(proba1 != 1.00){
						row = items[0] + "|" +url+"|" + String.valueOf(proba1)+"|"+String.valueOf(proba2);
						result = new Tuple2<String, String>(row, "");
					}
					return result;
				}
			}).filter(new Function<Tuple2<String, String>,Boolean>(){

				@Override
				public Boolean call(Tuple2<String, String> v1) throws Exception {
					// TODO Auto-generated method stub
					if(v1._1.isEmpty())
						return false;
					else
						return true;
				}
		    	
		    }).repartition(512);
			
		    s.saveAsHadoopFile(targetPath, Text.class, Text.class, TextOutputFormat.class);
		    ctx.stop();
		    return 0;
		 }
		 catch(Exception ex)
		 {
			 ex.printStackTrace();
			 return 1;
		 }
		 
		   
	   }
	
}
